package plugin

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/exec"
	"path/filepath"

	"github.com/hashicorp/go-hclog"
	goplugin "github.com/hashicorp/go-plugin"
	"github.com/louloulin/dataflare/pkg/message"
)

// RPCPluginImpl RPC插件实现
type RPCPluginImpl struct {
	id          string
	name        string
	version     string
	description string
	config      *PluginConfig
	client      *goplugin.Client
	started     bool
	process     interface{} // 实际的插件进程抽象
	status      PluginStatus
}

// NewRPCPluginImpl 创建新的RPC插件实现
func NewRPCPluginImpl(id, name, version, description string) *RPCPluginImpl {
	return &RPCPluginImpl{
		id:          id,
		name:        name,
		version:     version,
		description: description,
		config:      &PluginConfig{},
		started:     false,
		status:      StatusCreated,
	}
}

// ID 返回插件ID
func (p *RPCPluginImpl) ID() string {
	return p.id
}

// Name 返回插件名称
func (p *RPCPluginImpl) Name() string {
	return p.name
}

// Description 返回插件描述
func (p *RPCPluginImpl) Description() string {
	return p.description
}

// Type 返回插件类型
func (p *RPCPluginImpl) Type() PluginType {
	if p.config != nil && p.config.Type != "" {
		return p.config.Type
	}
	return PluginTypeProcessor
}

// Version 返回插件版本
func (p *RPCPluginImpl) Version() string {
	return p.version
}

// Status 返回插件状态
func (p *RPCPluginImpl) Status() PluginStatus {
	return p.status
}

// Config 返回插件配置
func (p *RPCPluginImpl) Config() *PluginConfig {
	return p.config
}

// Init 初始化插件
func (p *RPCPluginImpl) Init(ctx context.Context, config *PluginConfig) error {
	p.config = config
	p.status = StatusInitialized
	return nil
}

// Start 启动插件
func (p *RPCPluginImpl) Start(ctx context.Context) error {
	if p.started {
		return nil
	}

	// 获取插件路径
	pluginPath, ok := p.config.Config["plugin_path"].(string)
	if !ok {
		return fmt.Errorf("plugin_path not found in config")
	}

	// 确保插件文件存在
	if _, err := os.Stat(pluginPath); os.IsNotExist(err) {
		return fmt.Errorf("plugin file not found: %s", pluginPath)
	}

	// 使用hashicorp/go-plugin库启动RPC插件
	client := goplugin.NewClient(&goplugin.ClientConfig{
		HandshakeConfig: HandshakeConfig,
		Plugins:         PluginMap,
		Cmd:             exec.Command(pluginPath),
		AllowedProtocols: []goplugin.Protocol{
			goplugin.ProtocolNetRPC,
		},
		Logger: hclog.New(&hclog.LoggerOptions{
			Name:   "plugin-client",
			Level:  hclog.Info,
			Output: os.Stderr,
		}),
	})

	// 保存插件客户端
	p.client = client

	// 获取RPC客户端
	rpcClient, err := client.Client()
	if err != nil {
		p.status = StatusError
		return fmt.Errorf("failed to create client: %w", err)
	}

	// 确定插件类型
	pluginType := p.Type()

	fmt.Printf("Starting RPC plugin %s with type %s\n", p.id, pluginType)

	// 根据类型获取插件
	var raw interface{}
	switch pluginType {
	case PluginTypeProcessor:
		raw, err = rpcClient.Dispense("processor")
	case PluginTypeConnector:
		raw, err = rpcClient.Dispense("connector")
	default:
		p.status = StatusError
		return fmt.Errorf("unsupported plugin type: %s", pluginType)
	}

	if err != nil {
		p.status = StatusError
		return fmt.Errorf("failed to dispense plugin: %w", err)
	}

	p.process = raw
	p.started = true
	p.status = StatusRunning
	return nil
}

// Stop 停止插件
func (p *RPCPluginImpl) Stop(ctx context.Context) error {
	if !p.started {
		return nil
	}

	// 优雅地关闭插件
	if p.client != nil {
		p.client.Kill()
	}

	fmt.Printf("Stopping RPC plugin %s\n", p.id)
	p.started = false
	p.status = StatusStopped
	return nil
}

// Pause 暂停插件
func (p *RPCPluginImpl) Pause(ctx context.Context) error {
	if !p.started {
		return fmt.Errorf("plugin not started")
	}
	p.status = StatusPaused
	return nil
}

// Resume 恢复插件
func (p *RPCPluginImpl) Resume(ctx context.Context) error {
	if !p.started {
		return fmt.Errorf("plugin not started")
	}
	p.status = StatusRunning
	return nil
}

// Validate 验证插件配置
func (p *RPCPluginImpl) Validate(config *PluginConfig) error {
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	if config.ID == "" {
		return fmt.Errorf("plugin ID is required")
	}
	if config.Type == "" {
		return fmt.Errorf("plugin type is required")
	}
	return nil
}

// ProcessorRPCPlugin 处理器RPC插件实现
type ProcessorRPCPlugin struct {
	*RPCPluginImpl
}

// NewProcessorRPCPlugin 创建新的处理器RPC插件
func NewProcessorRPCPlugin(id, name, version, description string) *ProcessorRPCPlugin {
	return &ProcessorRPCPlugin{
		RPCPluginImpl: NewRPCPluginImpl(id, name, version, description),
	}
}

// Process 处理数据
func (p *ProcessorRPCPlugin) Process(ctx context.Context, input message.DataMessage) (message.DataMessage, error) {
	if !p.started {
		return nil, fmt.Errorf("plugin not started")
	}

	// 通过RPC调用插件的Process方法
	processor, ok := p.process.(interface{})
	if !ok {
		return nil, fmt.Errorf("plugin is not a valid processor interface")
	}

	// 调用Go plugin库返回的处理器接口上的Process方法
	// 这里我们不直接调用ProcessorPlugin接口，而是使用ProcessorRPCClient接口
	procClient, ok := processor.(*ProcessorRPCClient)
	if !ok {
		return nil, fmt.Errorf("plugin does not implement ProcessorRPCClient (type: %T)", processor)
	}

	return procClient.Process(ctx, input)
}

// ConnectorRPCPlugin 连接器RPC插件实现
type ConnectorRPCPlugin struct {
	*RPCPluginImpl
}

// NewConnectorRPCPlugin 创建新的连接器RPC插件
func NewConnectorRPCPlugin(id, name, version, description string) *ConnectorRPCPlugin {
	return &ConnectorRPCPlugin{
		RPCPluginImpl: NewRPCPluginImpl(id, name, version, description),
	}
}

// Connect 建立连接
func (p *ConnectorRPCPlugin) Connect(ctx context.Context) error {
	if !p.started {
		return fmt.Errorf("plugin not started")
	}

	// 通过RPC调用插件的Connect方法
	connector, ok := p.process.(ConnectorPlugin)
	if !ok {
		return fmt.Errorf("plugin is not a connector")
	}

	return connector.Connect(ctx)
}

// Disconnect 断开连接
func (p *ConnectorRPCPlugin) Disconnect(ctx context.Context) error {
	if !p.started {
		return fmt.Errorf("plugin not started")
	}

	// 通过RPC调用插件的Disconnect方法
	connector, ok := p.process.(ConnectorPlugin)
	if !ok {
		return fmt.Errorf("plugin is not a connector")
	}

	return connector.Disconnect(ctx)
}

// Send 发送数据
func (p *ConnectorRPCPlugin) Send(ctx context.Context, data interface{}) error {
	if !p.started {
		return fmt.Errorf("plugin not started")
	}

	// 通过RPC调用插件的Send方法
	connector, ok := p.process.(ConnectorPlugin)
	if !ok {
		return fmt.Errorf("plugin is not a connector")
	}

	return connector.Send(ctx, data)
}

// Receive 接收数据
func (p *ConnectorRPCPlugin) Receive(ctx context.Context) (interface{}, error) {
	if !p.started {
		return nil, fmt.Errorf("plugin not started")
	}

	// 通过RPC调用插件的Receive方法
	connector, ok := p.process.(ConnectorPlugin)
	if !ok {
		return nil, fmt.Errorf("plugin is not a connector")
	}

	return connector.Receive(ctx)
}

// RPCPluginManager RPC插件管理器
type RPCPluginManager struct {
	pluginsDir string
	plugins    map[string]*RPCPluginImpl
}

// NewRPCPluginManager 创建新的RPC插件管理器
func NewRPCPluginManager(pluginsDir string) *RPCPluginManager {
	return &RPCPluginManager{
		pluginsDir: pluginsDir,
		plugins:    make(map[string]*RPCPluginImpl),
	}
}

// LoadPlugin 加载RPC插件
func (m *RPCPluginManager) LoadPlugin(ctx context.Context, pluginPath string) (Plugin, error) {
	// 检查插件文件是否存在
	if _, err := os.Stat(pluginPath); os.IsNotExist(err) {
		return nil, fmt.Errorf("plugin file not found: %s", pluginPath)
	}

	// 获取插件ID（文件名）
	id := filepath.Base(pluginPath)
	ext := filepath.Ext(id)
	if ext != "" {
		id = id[:len(id)-len(ext)]
	}

	// 如果插件已经加载，返回已有实例
	if plugin, exists := m.plugins[id]; exists {
		return plugin, nil
	}

	// 创建插件配置
	config := map[string]interface{}{
		"plugin_path": pluginPath,
	}

	// 创建一个独立的描述命令来获取插件信息
	cmd := exec.Command(pluginPath, "--describe")
	output, err := cmd.Output()
	if err != nil {
		// 如果获取描述信息失败，尝试使用插件客户端
		var plugin Plugin
		plugin = NewRPCPluginImpl(id, id, "1.0.0", "RPC plugin "+id)

		// 初始化插件
		if err := plugin.Init(ctx, &PluginConfig{Config: config}); err != nil {
			return nil, fmt.Errorf("failed to initialize plugin: %w", err)
		}

		// 保存插件
		if p, ok := plugin.(*RPCPluginImpl); ok {
			m.plugins[id] = p
		}

		return plugin, nil
	}

	// 解析插件信息
	var info PluginInfo
	if err := json.Unmarshal(output, &info); err != nil {
		return nil, fmt.Errorf("failed to parse plugin info: %w", err)
	}

	// 根据插件类型创建相应的插件实例
	var plugin Plugin
	switch info.Type {
	case PluginTypeProcessor:
		plugin = NewProcessorRPCPlugin(info.ID, info.Name, info.Version, info.Description)
		config["plugin_type"] = string(PluginTypeProcessor)
	case PluginTypeConnector:
		plugin = NewConnectorRPCPlugin(info.ID, info.Name, info.Version, info.Description)
		config["plugin_type"] = string(PluginTypeConnector)
	default:
		plugin = NewRPCPluginImpl(info.ID, info.Name, info.Version, info.Description)
		config["plugin_type"] = string(info.Type)
	}

	// 初始化插件
	if err := plugin.Init(ctx, &PluginConfig{Config: config}); err != nil {
		return nil, fmt.Errorf("failed to initialize plugin: %w", err)
	}

	// 保存插件
	if p, ok := plugin.(*RPCPluginImpl); ok {
		m.plugins[id] = p
	}

	return plugin, nil
}

// LoadPlugins 加载指定目录下的所有RPC插件
func (m *RPCPluginManager) LoadPlugins(ctx context.Context) ([]Plugin, error) {
	// 确保插件目录存在
	if _, err := os.Stat(m.pluginsDir); os.IsNotExist(err) {
		log.Printf("Creating plugins directory: %s", m.pluginsDir)
		if err := os.MkdirAll(m.pluginsDir, 0755); err != nil {
			return nil, fmt.Errorf("failed to create plugins directory: %w", err)
		}
		return []Plugin{}, nil
	}

	// 读取插件目录
	files, err := os.ReadDir(m.pluginsDir)
	if err != nil {
		return nil, fmt.Errorf("failed to read plugins directory: %w", err)
	}

	var plugins []Plugin
	for _, file := range files {
		if file.IsDir() {
			continue
		}

		// 仅加载可执行文件（简化判断）
		filePath := filepath.Join(m.pluginsDir, file.Name())
		fileInfo, err := os.Stat(filePath)
		if err != nil {
			log.Printf("Failed to stat file %s: %v", filePath, err)
			continue
		}

		// 检查文件是否可执行
		if fileInfo.Mode()&0111 != 0 {
			plugin, err := m.LoadPlugin(ctx, filePath)
			if err != nil {
				log.Printf("Failed to load plugin %s: %v", filePath, err)
				continue
			}
			plugins = append(plugins, plugin)
		}
	}

	return plugins, nil
}

// CreateExampleRPCPlugin 创建示例RPC插件脚本
func CreateExampleRPCPlugin(pluginsDir string) error {
	// 确保插件目录存在
	if _, err := os.Stat(pluginsDir); os.IsNotExist(err) {
		if err := os.MkdirAll(pluginsDir, 0755); err != nil {
			return fmt.Errorf("failed to create plugins directory: %w", err)
		}
	}

	// 创建示例插件脚本
	scriptPath := filepath.Join(pluginsDir, "example-rpc-plugin.sh")
	scriptContent := `#!/bin/sh
if [ "$1" = "--describe" ]; then
    # 返回插件描述
    cat << EOF
{
    "id": "example-rpc-plugin",
    "name": "Example RPC Plugin",
    "type": "processor",
    "version": "1.0.0",
    "description": "An example RPC plugin for DataLink"
}
EOF
    exit 0
fi

# 启动插件服务器
echo "Starting example RPC plugin server..."
# 这里应该启动实际的RPC服务器
# 简化示例仅仅循环打印消息
while true; do
    echo "Example RPC plugin is running at $(date)"
    sleep 5
done
`

	// 写入文件
	if err := os.WriteFile(scriptPath, []byte(scriptContent), 0755); err != nil {
		return fmt.Errorf("failed to write plugin script: %w", err)
	}

	fmt.Printf("Created example RPC plugin script: %s\n", scriptPath)
	fmt.Println("In a real-world scenario, you would implement an actual plugin using hashicorp/go-plugin.")

	return nil
}
